package org.example.data_stream;

import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.List;

/**
 * @author shenguangyang
 */
public class E14_SinkToMySql {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        List<Event> events = new ArrayList<Event>() {{
            add(new Event("user02", "./cart", 2000L));
            add(new Event("user02", "./prod?id=10`", 3800L));
            add(new Event("user01", "./prod?id=15`", 4000L));
            add(new Event("user01", "./home1", 1000L));
            add(new Event("user02", "./prod?id=2`", 3000L));
            add(new Event("user03", "./prod?id=20`", 4000L));
            add(new Event("user04", "./prod?id=22`", 4000L));
            add(new Event("user05", "./prod?id=24`", 4000L));
        }};
        DataStream<Event> streamSource = env.fromCollection(events);

        streamSource.addSink(JdbcSink.sink(
                "insert into clicks (id, user, url, timestamp) values (?,?,?,?)",
                (ps, t) -> {
                    ps.setLong(1, t.getId());
                    ps.setString(2, t.getUser());
                    ps.setString(3, t.getUrl());
                    ps.setLong(4, t.getTimestamp());
                },
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:mysql://work01:53306/test?useUnicode=true&characterEncoding=utf8&characterSetResults=utf8&serverTimezone=Asia/Shanghai")
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        .withUsername("root")
                        .withPassword("project_dev@nwt")
                        .build())
        );
        env.execute();
    }
}
